package com.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.vertx.core.Vertx;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.Value;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.device.DeviceState;
import org.jetlinks.core.device.DeviceStateChecker;
import org.jetlinks.core.device.session.DeviceSessionManager;
import org.jetlinks.core.message.DeviceOfflineMessage;
import org.jetlinks.core.message.property.ReportPropertyMessage;
import org.jetlinks.supports.server.DecodedClientMessageHandler;
import org.springframework.util.ObjectUtils;
import reactor.core.publisher.Mono;

import javax.validation.constraints.NotNull;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;

/**
 * 定时设计状态检测器 </u>
 * 此方法会进入设备详情页面和刷新设备状态时候调用
 *
 */
@Slf4j
@AllArgsConstructor
public class TcpStateChecker implements DeviceStateChecker {
    private final Vertx vertx = Vertx.vertx();
    private DecodedClientMessageHandler handler;
    private DeviceSessionManager sessionManager;


    @Override
    public @NotNull Mono<Byte> checkState(@NotNull DeviceOperator device) {

        return device.getConfigs("host", "port")
            .flatMap(values -> {
                //从设备的配置文件中拿到host地址和端口号
                String host = values.getValue("host").map(Value::asString).orElse(null);
                int port = values.getValue("port").map(Value::asInt).orElse(0);
                //若不存在host地址和端口号则不再与设备创建连接
                if (ObjectUtils.isEmpty(host) || port == 0) {
                    return Mono.just(DeviceState.offline);
                }
                return sessionManager
                    .getSession(device.getDeviceId())
                    //判断session是否存在，存在则表明已与设备建立连接，不存在则表明需要与设备建立连接
                    .switchIfEmpty(Mono.create((sink) -> vertx
                        //通过host地址和端口号和建立连接
                        .createNetClient()
                        .connect(port, host, async -> {
                            if (async.succeeded()) {
                                log.info("connect to {}:{} success", host, port);
                                //若连接成功，则创建session并把连接放入其中
                                TcpClientSession newSession = new TcpClientSession(device.getDeviceId(), device, async.result());
                                //设备发送消息时，会触发async.result().handler(...)
                                async.result().handler(buffer -> {
                                    //将设备发送的消息转为byte数组，再根据实际的设备文档进行按位解析
                                    log.info("receive message:{}", buffer.toString());
                                    byte[] bytes = buffer.getBytes();
                                    ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
                                    //根据收到的报文解析成平台对应的消息类型，此处示例封装为上报属性消息
                                    HashMap<String, Object> reportMap = doDecode(byteBuf);
                                    //封装设备上报消息
                                    ReportPropertyMessage rpm = new ReportPropertyMessage();
                                    rpm.setDeviceId(device.getDeviceId());
                                    rpm.setProperties(reportMap);
                                    //将封装的消息交给消息处理器处理
                                    handler.handleMessage(device, rpm)
                                        .subscribe();
                                });
                                //设备主动断开连接时，会触发async.result().closeHandler(...)
                                async.result().closeHandler(close -> {
                                    //关闭session
                                    newSession.close();
                                    //清除session
                                    sessionManager.remove(device.getDeviceId(), false);
                                    //封装设备离线消息
                                    DeviceOfflineMessage off = new DeviceOfflineMessage();
                                    off.setDeviceId(device.getDeviceId());
                                    //将封装的消息交给消息处理器处理
                                    handler.handleMessage(device, off)
                                        .subscribe();
                                });
                                if(async.failed()){
                                    log.error("connect to {}:{} failed", host, port);
                                    System.out.println(String.format("connect to {%s}:{%s} failed",host, port));
                                    sink.error(async.cause());
                                    sessionManager.remove(device.getDeviceId(), false).thenReturn(DeviceState.offline);
                                    return ;
                                }



                                //将新建的session放入流中
                                sink.success(newSession);
                            } else {
                                sink.error(async.cause());
                            }
                        })
                    ))
                    //将session存入session对象中
                    .flatMap(session -> sessionManager
                        .compute(device.getDeviceId(), old -> old
                            .map(sess -> session)
                            //会话已存在则替换为新的会话
                            .defaultIfEmpty(session))
                    )

                    //使设备实例上线
                    .thenReturn(DeviceState.online)
                    //出现错误时，清除设备session并使设备离线
                    .onErrorResume(err -> sessionManager.remove(device.getDeviceId(), false)
                        .thenReturn(DeviceState.offline));
            });
    }

    private HashMap<String, Object> doDecode(ByteBuf byteBuf) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("wendu", byteBuf.readDouble());
        return map;

    }
}
